/*
 * Singularity Engine
 * Copyright (C) 2010 Vyacheslav Vorobyov, All rights reserved.
 * See LICENSE for full license information.
 */

/**
 * 
 */
package org.singularity.mapred.framework;

import java.io.IOException;
import java.util.List;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.singularity.io.CloseablePipe;
import org.singularity.io.Pipe;

/**
 * @author vjache
 *
 */
public class HashedDemultiplexerPipe<T> extends DemultiplexerPipe<T> {
	
	private static Log LOG = LogFactory.getFactory().getInstance(HashedDemultiplexerPipe.class);

	private final List<CloseablePipe<T>> _pipes;

	public HashedDemultiplexerPipe(List<CloseablePipe<T>> aPipes)
	{
		_pipes = aPipes;
	}
	@Override
	protected CloseablePipe<T> selectFor(T aItem) {
		int i = Math.abs(aItem.hashCode()) % _pipes.size();
		return _pipes.get(i);
	}

	@Override
	public void flush() throws IOException {
		for(Pipe<?> p:_pipes)
		{
			p.flush();
		}
	}
	@Override
	public void close() throws IOException {
		flush();
		for(CloseablePipe<T> p:_pipes)
		{
			try{p.close();} catch(IOException e){
				LOG.error(String.format("Error while closing %s pipe.", p), e);
			}
		}
	}
}
